package com.frinder.hadoop.hbase.original;

import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Iterator;
import java.util.List;

/**
 * @author frinder
 * @date 2018/6/30
 * @Description: 实际业务类，实现 hbase 各种操作
 */
@Slf4j
public class HBaseHelper {

    private Admin getAdmin(Connection conn) throws Exception {
        return conn.getAdmin();
    }

    /**
     * 创建表
     *
     * @param conn
     * @param tableName
     * @param cols
     * @throws Exception
     */
    public void createTable(Connection conn, String tableName, String... cols) throws Exception {
        log.info("*** begin create table: {}, cols: {}", tableName, Joiner.on(",").join(cols));
        try (Admin admin = getAdmin(conn)) {
            TableName table = TableName.valueOf(tableName);
            if (admin.tableExists(table)) {
                log.warn("*** table {} already exists!", tableName);
            } else {
                HTableDescriptor tableDescriptor = new HTableDescriptor(table);
                for (String col : cols) {
                    tableDescriptor.addFamily(new HColumnDescriptor(col));
                }
                admin.createTable(tableDescriptor);
                log.info("*** finish create table: {}, cols: {}", tableName, Joiner.on(",").join(cols));
            }
        }
    }

    /**
     * 删除表
     *
     * @param conn
     * @param tableName
     * @throws Exception
     */
    public void deleteTable(Connection conn, String tableName) throws Exception {
        log.info("*** begin delete table: {}", tableName);
        try (Admin admin = getAdmin(conn)) {
            TableName table = TableName.valueOf(tableName);
            if (admin.tableExists(table)) {
                admin.disableTable(table);
                admin.deleteTable(table);
                log.info("*** finish delete table: {}", tableName);
            }
        }
    }

    /**
     * 查询所有表
     *
     * @param conn
     * @return
     * @throws Exception
     */
    public List<TableName> getTables(Connection conn) throws Exception {
        log.info("*** begin get tables");
        try (Admin admin = getAdmin(conn)) {
            TableName[] tableNames = admin.listTableNames();
            log.info("*** finish get tables: {}", Joiner.on(",").join(tableNames));
            return Lists.newArrayList(tableNames);
        }
    }

    /**
     * @param conn
     * @param tableName 表名
     * @param rowKey    行标识
     * @param colFamily 簇名
     * @param col       列名
     * @param val       值
     */
    public void insertRow(Connection conn, String tableName, String rowKey, String colFamily, String col, String val) throws Exception {
        log.info("*** begin insert table: {}, rowKey: {}, colFamily: {}, col: {}, val: {}",
                tableName,
                rowKey,
                colFamily,
                col,
                val
        );
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col), Bytes.toBytes(val));
            table.put(put);
            log.info("*** finish insert table: {}, rowKey: {}, colFamily: {}, col: {}, val: {}",
                    tableName,
                    rowKey,
                    colFamily,
                    col,
                    val
            );
        }
    }

    /**
     * 删除记录
     *
     * @param conn
     * @param tableName
     * @param rowKey
     * @throws Exception
     */
    public void deleteRow(Connection conn, String tableName, String rowKey) throws Exception {
        log.info("*** begin delete table: {}, rowKey: {}",
                tableName,
                rowKey
        );
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            log.info("*** finish delete table: {}, rowKey: {}",
                    tableName,
                    rowKey
            );
        }
    }

    /**
     * 删除 簇列
     *
     * @param conn
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param col
     * @throws Exception
     */
    public void deleteCol(Connection conn, String tableName, String rowKey, String colFamily, String col) throws Exception {
        log.info("*** begin delete table: {}, rowKey: {}, colFamily: {}, col: {}",
                tableName,
                rowKey,
                colFamily,
                col
        );
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addFamily(Bytes.toBytes(colFamily));
            delete.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
            table.delete(delete);
            log.info("*** finish delete table: {}, rowKey: {}, colFamily: {}, col: {}",
                    tableName,
                    rowKey,
                    colFamily,
                    col
            );
        }
    }

    /**
     * 获取记录
     *
     * @param conn
     * @param tableName
     * @param rowKey
     * @return
     * @throws Exception
     */
    public Result getRowData(Connection conn, String tableName, String rowKey) throws Exception {
        log.info("*** begin get table: {}, rowKey: {}",
                tableName,
                rowKey
        );
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            formatResult(result);
            return result;
        }
    }

    /**
     * 获取 簇列
     *
     * @param conn
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param col
     * @return
     * @throws Exception
     */
    public Result getColData(Connection conn, String tableName, String rowKey, String colFamily, String col) throws Exception {
        log.info("*** begin get table: {}, rowKey: {}, colFamily: {}, col: {}",
                tableName,
                rowKey,
                colFamily,
                col
        );
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addFamily(Bytes.toBytes(colFamily));
            get.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
            Result result = table.get(get);
            formatResult(result);
            return result;
        }
    }

    /**
     * 模式化结果
     *
     * @param result
     */
    public void formatResult(Result result) {
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            log.info("*** RowName: {}, Timestamp: {}, ColFamily: {}, Col: {}, Val: {}",
                    new String(CellUtil.cloneRow(cell)),
                    cell.getTimestamp(),
                    new String(CellUtil.cloneFamily(cell)),
                    new String(CellUtil.cloneQualifier(cell)),
                    new String(CellUtil.cloneValue(cell))
            );
        }
    }

    /**
     * 批量查看记录
     *
     * @param conn
     * @param tableName
     * @param startRow
     * @param stopRow
     * @return
     * @throws Exception
     */
    public List<Result> scanData(Connection conn, String tableName, String startRow, String stopRow) throws Exception {
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRow));
            scan.setStopRow(Bytes.toBytes(stopRow));
            ResultScanner scanner = table.getScanner(scan);
            Iterator<Result> it = scanner.iterator();
            List<Result> resultList = Lists.newArrayList();
            while (it.hasNext()) {
                Result result = it.next();
                formatResult(result);
                resultList.add(result);
            }
            return resultList;
        }
    }

}
